package spark.MLlib;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.classification.SVMModel;
import org.apache.spark.mllib.classification.SVMWithSGD;
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;
import scala.Tuple2;

/**
 * 作者: LDL
 * 功能说明:
 * 创建日期: 2015/6/30 15:29
 */
public class SVMClassifier {

    public static void main(String[] args) {
        System.setProperty("hadoop.home.dir", "F:\\tools\\hadoop-common-2.2.0-bin-master");
        SparkConf conf = new SparkConf().setMaster("local").setAppName("JavaDataTypes");
        JavaSparkContext jsc = new JavaSparkContext(conf);

        /*JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotrans\\libsvm.txt").toJavaRDD();
        JavaRDD<LabeledPoint> training = data.sample(false, 0.6, 15L);

        training.cache();
        JavaRDD<LabeledPoint> test = data.subtract(training);*/
        //jsc.setLogLevel("OFF");
        JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotrans\\libsvm.txt").toJavaRDD();
        JavaRDD<LabeledPoint> test = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotest\\libsvm.txt").toJavaRDD();
        data.cache();
        // Run training algorithm to build the model.
        int numIterations = 100;
        final SVMModel model = SVMWithSGD.train(data.rdd(), numIterations);
        //SVMModel model = SVMModel.load(jsc.sc(), "D:/test.txt");
        //Clear the default threshold.
        model.clearThreshold();

        // Compute raw scores on the test set.
        JavaRDD<Tuple2<Object, Object>> scoreAndLabels = test.map(
                p -> {
                    Double score = model.predict(p.features());
                    return new Tuple2<>(score, p.label());
                }
        );
        // Get evaluation metrics.
        BinaryClassificationMetrics metrics =
                new BinaryClassificationMetrics(JavaRDD.toRDD(scoreAndLabels));
        double auROC = metrics.areaUnderPR();

        System.out.println("Area under ROC = " + auROC);
    }
}
